package cn.tedu.charging.order.amqp;

import cn.tedu.charging.common.constant.AmqpComponentConst;
import cn.tedu.charging.common.pojo.message.DelayCheckMessage;
import cn.tedu.charging.common.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.order.mapper.BillFailMapper;
import cn.tedu.charging.order.mapper.BillSuccessMapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class AmqpConsumer {
    //对接死信队列 消费延迟消息DelayCheckMessage
    @Autowired
    private BillSuccessMapper billSuccessMapper;
    @Autowired
    private BillFailMapper billFailMapper;
    @Autowired
    private RedisTemplate redisTemplate;
    @RabbitListener(queues = AmqpComponentConst.DELAY_BILL_DL_QUEUE)
    public void consumeDelayBill(DelayCheckMessage msg, Message message, Channel channel){
        log.debug("消费到延迟消息:{}",msg);
        //1.抢锁 setnx key value ex 5
        String lockKey="charging:bill:lock:"+msg.getOrderNo();
        String value="";
        ValueOperations stringOps = redisTemplate.opsForValue();
        try {
            //dowhile 循环 直到抢到锁为止
            boolean haveLock=false;
            int count=0;
            do {
                count++;
                if(count>1){
                    //不是第一次抢锁
                    Thread.sleep(1000);
                }
                haveLock=stringOps.setIfAbsent(lockKey,value,5, TimeUnit.SECONDS);
            }while(!haveLock);
            //具体消费逻辑
            boolean ack=false;
            handleDelayBill(msg);//消费核心
            //手动确认 消费完毕
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            ack=true;
        } catch (Exception e) {
            //处理延迟消息出现异常 手动确认重新发送消息 重新处理3次以上 拒绝处理 人工补偿
            log.error("确认消息失败");
        }finally {
            //无论当前业务是否执行成功 都需要释放锁
            redisTemplate.delete(lockKey);
        }
    }

    private void handleDelayBill(DelayCheckMessage msg) {
        //1.查询成功 和失败 是否存在
        QueryWrapper queryWrapper=new QueryWrapper();
        queryWrapper.eq("bill_id",msg.getOrderNo());
        Long exSuccess=billSuccessMapper.selectCount(queryWrapper);
        Long exFail=billFailMapper.selectCount(queryWrapper);
        if (exSuccess>0l||exFail>0l){
            //2.判断如果存在结束
            return;
        }else{
            //3.如果不存在则进入失败处理流程 设备无响应
            ChargingBillFailPO po=new ChargingBillFailPO();
            //订单编号 用户id 时间
            po.setBillId(msg.getOrderNo());
            po.setUserId(msg.getUserId());
            po.setCreateTime(new Date());
            po.setUpdateTime(new Date());
            po.setFailDesc("设备无响应");
            billFailMapper.insert(po);
            //UNDO 设计一个修改订单状态的device服务接口
            //TODO 实时消息推送 告知设备故障 或者其他原因 订单无法继续充电 请换枪 赠送优惠券
        }

    }
}
